package com.nba;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class ChampionCount {
    //建立任务
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //第一步：建立job对象
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);//建立作业
        //第二步：设置驱动类
        job.setJarByClass(ChampionCount.class);  //绑定作业的类
        job.setJobName("ChampionCount"); //设置作业名称
        //第三步：设置Mapper类
        job.setMapperClass(ChampionCount.MyMapper.class); //设置Mapper类
        job.setMapOutputKeyClass(Text.class); //设置Map的输出的key的类型
        job.setMapOutputValueClass(IntWritable.class); //设置Map的输出的value类型

        //第四步：设置Reducer类
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setCombinerClass(MyReducer.class);
        //设置分区
        job.setPartitionerClass(MyPartitioner.class); //设置使用自定义的分区文件
        job.setNumReduceTasks(3);//设置分区任务数量

        //第五步:设置输入和输出文件的路径
        // hadoop jar ....jar ..类名 分析的源文件的路径  输出的结果文件的路径
        FileInputFormat.setInputPaths(job, new Path("file/NBA/dataclear/out/part-m-00000"));  //设置要分析的源文件路径
        FileOutputFormat.setOutputPath(job, new Path("file/NBA/ChampionCount/out")); //设置要输出的结果文件路径

        //第六步：提交程序
        job.waitForCompletion(true);
    }

    /**
     * mapper组件
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        /**
         * 完成 冠军队的数量统计 如key:冠军队名称,E/W/无 value:1
         *
         * @param key
         * @param value
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            //value:如：1970,4.24-5.8,纽约尼克斯队,4-3,洛杉矶湖人队,威利斯·里德,E
            String[] split = value.toString().split(",");
            //取出冠军队的名称
            String champion = split[2];
            String area = null;//获取分区标志
            if (split.length == 7) { //带有分区标志的
                area = split[6]; //得到分区标志
            }
            Text newKey = new Text();
            if (area != null) {
                newKey.set(champion + "," + area); //有分区标志的，key如纽约尼克斯队,E
            } else {
                newKey.set(champion);//无分区标志
            }

            context.write(newKey, new IntWritable(1));  //如：写入的就是：如纽约尼克斯队,E    1
        }
    }

    /**
     * Reducer组件
     */
    static class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        /**
         * 针对冠军队的数量进行统计
         *
         * @param key     如：约尼克斯队,E
         * @param values  如：[1,1,1,1,1]
         * @param context 输出的： 如：key：约尼克斯队,E   value： 5
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable value : values) {
                sum += value.get();
            }
            context.write(key, new IntWritable(sum));
        }
    }

    //自定义的分区组件：规则按照分区标志W E 无分区进行分别存放
    static class MyPartitioner extends Partitioner<Text, IntWritable> {

        /**
         * 进行分区的文件
         *
         * @param key   三种样例数据：纽约尼克斯队,E  洛杉矶湖人队,W  波士顿凯尔特人队
         * @param value
         * @param i
         * @return
         */
        @Override
        public int getPartition(Text key, IntWritable value, int i) {
            String[] split = key.toString().split(",");
            String area = null;
            if (split.length > 1) {//带分区标志
                area = split[1];  //得到分区的标记
            }
            //根据area的值进行分区
            if ("E".equals(area)) {
                return 0;  //如果area=E，编号到part-r-00000文件中
            } else if ("W".equals(area)) {
                return 1;  //如果area=W，编号到part-r-00001文件中
            } else {
                return 2;  //如果area4=null，编号到part-r-00002文件中
            }
        }
    }
}
